package com.archgeek.java.queuedrain;

import com.google.common.collect.Queues;
import org.apache.commons.lang3.RandomUtils;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @author pizhihui
 * @date 2023-04-26 11:16
 */
public class QueueDrainDemo {


    ArrayBlockingQueue<Integer> queue = new java.util.concurrent.ArrayBlockingQueue(1000);


    ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(5);




    public void doRun() throws InterruptedException {

        scheduledExecutorService.scheduleAtFixedRate(() -> {

            List<Integer> list = new ArrayList<>();
            Iterator<Integer> iterator = queue.iterator();
            while (iterator.hasNext()) {
                Integer next = iterator.next();
                list.add(next);
            }


            System.out.println(Thread.currentThread().getName() + "定时调度线程定期查看数据: " + list.size() + ",queue.size()=" + queue.size());
        }, 0, 10, TimeUnit.SECONDS);

        new Thread(() -> {
            while (true) {
                try {
                    HashSet<Integer> hashSet = new HashSet<>();
                    System.out.println(Thread.currentThread().getName() + "等待取数.....");
                    Queues.drain(queue, hashSet, 1, 1, TimeUnit.MINUTES);
                    System.out.println(Thread.currentThread().getName() + "取到数据了....." + hashSet.size());
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }

        }).start();



        TimeUnit.MILLISECONDS.sleep(1000);

        new Thread(() -> {
            for (int i = 0; i < 100; i++) {
                System.out.println(Thread.currentThread().getName() + "放入数据.....");
                queue.offer(i);
                try {
                    TimeUnit.SECONDS.sleep(RandomUtils.nextInt(0,10));
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }

        }).start();


    }


    public static void main(String[] args) {


        try {
            new QueueDrainDemo().doRun();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }



    }


}
